/*
异步工作器
*/

#ifndef LOOPER_HPP
#define LOOPER_HPP

#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>
#include <atomic>

namespace xzt
{
    namespace log
    {
        using Functor = std::function<void(Buffer::ptr&)>;

        enum class AsyncSafeType
        {
            ASYNC_SAFE,  // 安全状态，表示缓冲区满了就等待消费者消费，避免资源耗尽的风险
            ASYNC_UNSAFE  // 不安全状态，无限扩容，用于测试
        };

        class AsyncLooper
        {
        public:
            using ptr = std::shared_ptr<AsyncLooper>;

            AsyncLooper(const Functor& cb, AsyncSafeType type = AsyncSafeType::ASYNC_SAFE): 
                _looper_type(type), 
                _pro_buffer(std::make_shared<Buffer>()), 
                _con_buffer(std::make_shared<Buffer>()), 
                _is_stop(false), 
                _worker_thread(std::thread(&AsyncLooper::threadEntry, this)),
                _callback(cb) {}
            ~AsyncLooper() { stop(); }

            void stop()
            {
                _is_stop = true;
                _cv_con.notify_all();  // 唤醒所有等待消费者的线程
                _worker_thread.join();  // 等待工作线程退出
            }

            void push(const char* data, size_t len)
            {
                // 生产者向缓冲区中写入数据
                std::unique_lock<std::mutex> lock(_mtx);
                // 1.无限扩容（非安全）  2.剩余可写入空间小于写入数据长度时，等待消费者消费(安全)
                // 利用条件变量控制，当缓冲区内空间大于等于写入数据长度时，则可以添加数据，否则等待消费者消费
                if(_looper_type == AsyncSafeType::ASYNC_SAFE)
                {
                    _cv_pro.wait(lock, [&]()->bool { return _con_buffer->writeAbleSize() >= len; });
                }
                
                // 走到这里说明有足够空间可以添加数据
                _pro_buffer->push(data, len);
                // 通知消费者消费数据
                _cv_con.notify_one();
            }

        private:
            void threadEntry()  // 异步工作器的工作线程入口函数
            {
                while(1)  // 循环等待停止信号
                {
                    {
                        std::unique_lock<std::mutex> lock(_mtx);
                        // 退出标志为真，且生产者缓冲区为空，则退出循环(否则存在退出标志为真，但剩余日志信息未落地的情况)
                        if(_is_stop && _pro_buffer->empty())
                        {
                            break;
                        }
                        // 为互斥锁设定作用域，当缓冲区交换完毕后就释放互斥锁
                        // 1.若缓冲区有数据，则进行交换，无则阻塞等待生产者生产数据
                        _cv_con.wait(lock, [&]() -> bool 
                                { return _is_stop || !_pro_buffer->empty(); });

                        // 进行缓冲区交换
                        _con_buffer.swap(_pro_buffer);
                        // 2.通知生产者生产数据
                        if(_looper_type == AsyncSafeType::ASYNC_SAFE)
                        {
                            _cv_pro.notify_all();
                        }
                    }
                    // 3.处理消费者者缓冲区中的数据
                    _callback(_con_buffer);  // 调用回调函数处理
                    // 4.重置消费者缓冲区
                    _con_buffer->reset();
                }
            }
        private:
            Functor _callback;  // 异步工作器的回调函数, 用于处理生产者缓冲区中的数据

        private:
            AsyncSafeType _looper_type;  // 异步工作器的安全类型
            Buffer::ptr _pro_buffer;  // 生产者缓冲区
            Buffer::ptr _con_buffer;  // 消费者缓冲区
            std::mutex _mtx;
            std::condition_variable _cv_pro;  // 生产者条件变量
            std::condition_variable _cv_con;  // 消费者条件变量
            std::atomic<bool> _is_stop;  // 停止标志
            std::thread _worker_thread;  // 异步工作器对应的工作线程
        };
    }
}




#endif // LOOPER_HPP